package com.example.yckjbigdataservice.controller;

import com.example.yckjbigdataservice.config.HbaseConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.hindex.client.HIndexAdmin;
import org.apache.hadoop.hbase.hindex.client.impl.HIndexClient;
import org.apache.hadoop.hbase.hindex.common.HIndexSpecification;
import org.apache.hadoop.hbase.hindex.common.TableIndices;
import org.apache.hadoop.hbase.hindex.protobuf.generated.HIndexProtos;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.regionserver.BloomType;
import org.apache.hadoop.hbase.security.access.AccessControlClient;
import org.apache.hadoop.hbase.security.access.Permission;
import org.apache.hadoop.hbase.security.access.PermissionStorage;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.context.ApplicationContext;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;
import java.net.URLDecoder;
import java.util.*;
import java.util.concurrent.CompletableFuture;

@RequestMapping("/hbase")
@RestController
@Slf4j
public class HBaseController {

    private final static TableName tableName = TableName.valueOf("hbase_sample_table");

    private final Connection hbaseConnection;

    //private final ApplicationContext context;

    public HBaseController(ApplicationContext context) {
        boolean isExs = context.containsBean("createHbaseConfiguration");
        try {
            if(!isExs){
                this.hbaseConnection = null;
            }else {
                this.hbaseConnection = ConnectionFactory.createConnection(context.getBean("createHbaseConfiguration",Configuration.class));
            }
            //this.hbaseConnection = ConnectionFactory.createConnection(createHbaseConfiguration);
            //this.hbaseConnection = HbaseConfig.getHbaseConnection(createHbaseConfiguration);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @GetMapping("/data/scan")
    public String scanData(@RequestParam("tableName") String tableName,
                           @RequestParam("family") String family,
                           @RequestParam("qualifier") String qualifier,
                           @RequestParam("value") String value) {
        log.info("开始扫描数据");

        Table table = null;
        ResultScanner scanner = null;
        try {
            //table = hbaseConnection.getTable(TableName.valueOf(URLDecoder.decode(tableName, "UTF-8")));
            table = hbaseConnection.getTable(TableName.valueOf("std:" + tableName));

            // Create a filter for indexed column.
            Filter filter = new SingleColumnValueFilter(Bytes.toBytes(family), Bytes.toBytes(qualifier),
                    CompareOperator.EQUAL,value.getBytes());
            Scan scan = new Scan();
            scan.setFilter(filter);
            scanner = table.getScanner(scan);
            for (Result result : scanner) {
                for (Cell cell : result.rawCells()) {
                    log.info("{}:{},{},{}", Bytes.toString(CellUtil.cloneRow(cell)),
                            Bytes.toString(CellUtil.cloneFamily(cell)), Bytes.toString(CellUtil.cloneQualifier(cell)),
                            Bytes.toString(CellUtil.cloneValue(cell)));
                }
            }
            log.info("扫描数据完成");
        } catch (IOException e) {
            log.error("Scan data by index failed ", e);
        } finally {
            if (scanner != null) {
                // Close the scanner object.
                scanner.close();
            }
            try {
                if (table != null) {
                    table.close();
                }
            } catch (IOException e) {
                log.error("Close table failed ", e);
            }
        }
        log.info("Exiting testScanDataByIndex.");
        return "数据扫描success";
    }


    // 原子操作
    public static void main1(String[] args) throws Exception {
        Connection connection = ConnectionFactory.createConnection();
        TableName tableName = TableName.valueOf("mytable");
        Table table = connection.getTable(tableName);

        byte[] rowKey = Bytes.toBytes("row1");
        byte[] family = Bytes.toBytes("cf");
        byte[] qualifier = Bytes.toBytes("col1");
        byte[] value = Bytes.toBytes("value1");

        // 创建一个原子操作
        BufferedMutatorParams params = new BufferedMutatorParams(tableName);
        params.writeBufferSize(1024 * 1024);
        BufferedMutator mutator = connection.getBufferedMutator(params);

        // 增加一个Put操作
        Put put = new Put(rowKey);
        put.addColumn(family, qualifier, value);
        mutator.mutate(put);

        // 增加一个Delete操作
        Delete delete = new Delete(rowKey);
        delete.addColumn(family, qualifier);
        mutator.mutate(delete);

        // 执行原子操作
        mutator.flush();

        // 关闭连接
        mutator.close();
        table.close();
        connection.close();
    }

    // 异步操作
    public static void main2(String[] args) throws Exception {
        AsyncConnection connection = ConnectionFactory.createAsyncConnection().get();
        TableName tableName = TableName.valueOf("mytable");
        AsyncTable<AdvancedScanResultConsumer> asyncTable = connection.getTable(tableName);

        byte[] rowKey = Bytes.toBytes("row1");

        // 异步执行Get操作
        CompletableFuture<Result> getFuture = asyncTable.get(new Get(rowKey));
        getFuture.thenAccept(result -> {
            // 处理Get操作的结果
            List<Cell> cells = result.listCells();
            if (cells != null) {
                for (Cell cell : cells) {
                    byte[] value = cell.getValueArray();
                    // 处理值
                }
            }
        });

        // 异步执行Put操作
        Put put = new Put(rowKey);
        put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("col1"), Bytes.toBytes("value1"));
        CompletableFuture<Void> putFuture = asyncTable.put(put);
        putFuture.thenRun(() -> {
            // Put操作完成后的处理
        });

        // 等待所有异步操作完成
        CompletableFuture.allOf(getFuture, putFuture).join();

        // 关闭连接
        //asyncTable.;
        connection.close();
    }



    /**
     * HBaseSample test
     */
    public void test() {
        try {
            testCreateTable();
            testMultiSplit();
            testPut();
            createIndex();
            enableIndex();
            testScanDataByIndex();
            testModifyTable();
            testGet();
            testScanData();
            testSingleColumnValueFilter();
            testFilterList();
            testDelete();
            disableIndex();
            dropIndex();
            dropTable();
        } finally {
            if (hbaseConnection != null) {
                try {
                    hbaseConnection.close();
                } catch (IOException e1) {
                    log.error("Failed to close the connection ", e1);
                }
            }
        }
    }

    /**
     * Create user info table
     */
    public void testCreateTable() {
        log.info("Entering testCreateTable.");

        // Specify the table descriptor.
        TableDescriptorBuilder htd = TableDescriptorBuilder.newBuilder(tableName);

        // Set the column family name to info.
        ColumnFamilyDescriptorBuilder hcd = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("info"));

        // Set data encoding methods. HBase provides DIFF,FAST_DIFF,PREFIX
        hcd.setDataBlockEncoding(DataBlockEncoding.FAST_DIFF);

        // Set compression methods, HBase provides two default compression
        // methods:GZ and SNAPPY
        // GZ has the highest compression rate,but low compression and
        // decompression effeciency,fit for cold data
        // SNAPPY has low compression rate, but high compression and
        // decompression effeciency,fit for hot data.
        // it is advised to use SANPPY
        // GZIP : 优点-提供最高的压缩率，适合对存储空间要求高的场景; 缺点-压缩和解压缩速度较慢，对CPU资源消耗较大
        // SNAPPY : 优点-压缩和解压缩速度快，适合对性能要求高的场景; 缺点-压缩率相对较低，适合对存储空间要求不高的场景
        // LZO : 优点-提供较高的压缩率和较快的压缩速度，适合综合性能要求的场景; 缺点-需要额外安装LZO库，不同平台的兼容性可能有所差异
        // LZ4 : 优点-提供非常高的压缩和解压缩速度，适合对延迟敏感的场景; 缺点-压缩率相对较低，适合对存储空间要求不高的场景
        hcd.setCompressionType(Compression.Algorithm.SNAPPY);
        hcd.setBloomFilterType(BloomType.ROW);

        htd.setColumnFamily(hcd.build());

        Admin admin = null;
        try {
            // Instantiate an Admin object.
            admin = hbaseConnection.getAdmin();
            if (!admin.tableExists(tableName)) {
                log.info("Creating table...");
                //admin.createTable(htd.build());

                // 定义预分区的行键范围
                byte[][] splitKeys = new byte[][] {
                        Bytes.toBytes("region1"),
                        Bytes.toBytes("region2"),
                        Bytes.toBytes("region3")
                };
                admin.createTable(htd.build(), splitKeys);

                log.info(admin.getClusterMetrics().toString());
                log.info(admin.listNamespaceDescriptors().toString());
                log.info("Table created successfully.");
            } else {
                log.warn("table already exists");
            }
        } catch (IOException e) {
            log.error("Create table failed.", e);
        } finally {
            if (admin != null) {
                try {
                    // Close the Admin object.
                    admin.close();
                } catch (IOException e) {
                    log.error("Failed to close admin ", e);
                }
            }
        }
        log.info("Exiting testCreateTable.");
    }

    /**
     * 修改压缩算法
     */
    public static void xiugaiYasu() {
        Configuration config = HBaseConfiguration.create();
        try (Connection connection = ConnectionFactory.createConnection(config);
             Admin admin = connection.getAdmin()) {

            // 获取表名
            TableName tableName = TableName.valueOf("compressed_table");

            // 获取当前的表描述符
            TableDescriptor tableDescriptor = admin.getDescriptor(tableName);

            // 获取当前的列族描述符，并设置新的压缩算法为SNAPPY
            ColumnFamilyDescriptor columnFamilyDescriptor = tableDescriptor.getColumnFamily("data".getBytes());
            ColumnFamilyDescriptor newColumnFamilyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(columnFamilyDescriptor)
                    .setCompressionType(Compression.Algorithm.SNAPPY)
                    .build();

            // 使用新的列族描述符重新构建表描述符
            TableDescriptor newTableDescriptor = TableDescriptorBuilder.newBuilder(tableDescriptor)
                    .modifyColumnFamily(newColumnFamilyDescriptor)
                    .build();

            // 更新表的压缩算法
            admin.modifyTable(newTableDescriptor);
            System.out.println("Table compression algorithm changed to SNAPPY.");

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    /**
     * testMultiSplit
     */
    public void testMultiSplit() {
        log.info("Entering testMultiSplit.");

        Table table = null;
        Admin admin = null;
        try {
            admin = hbaseConnection.getAdmin();

            // initilize a HTable object
            table = hbaseConnection.getTable(tableName);
            Set<RegionInfo> regionSet = new HashSet<RegionInfo>();
            List<HRegionLocation> regionList = hbaseConnection.getRegionLocator(tableName).getAllRegionLocations();
            for (HRegionLocation hrl : regionList) {
                regionSet.add(hrl.getRegion());
            }
            byte[][] sk = new byte[4][];
            sk[0] = "A".getBytes();
            sk[1] = "D".getBytes();
            sk[2] = "F".getBytes();
            sk[3] = "H".getBytes();
            for (RegionInfo regionInfo : regionSet) {
                admin.multiSplitSync(regionInfo.getRegionName(), sk);
            }
            log.info("MultiSplit successfully.");
        } catch (IOException e) {
            log.error("MultiSplit failed ", e);
        } finally {
            if (table != null) {
                try {
                    // Close table object
                    table.close();
                } catch (IOException e) {
                    log.error("Close table failed ", e);
                }
            }
            if (admin != null) {
                try {
                    // Close the Admin object.
                    admin.close();
                } catch (IOException e) {
                    log.error("Close admin failed ", e);
                }
            }
        }
        log.info("Exiting testMultiSplit.");
    }

    private static Put putData(byte[] familyName, byte[][] qualifiers, List<String> data) {
        Put put = new Put(Bytes.toBytes(data.get(0)));
        put.addColumn(familyName, qualifiers[0], Bytes.toBytes(data.get(1)));
        put.addColumn(familyName, qualifiers[1], Bytes.toBytes(data.get(2)));
        put.addColumn(familyName, qualifiers[2], Bytes.toBytes(data.get(3)));
        put.addColumn(familyName, qualifiers[3], Bytes.toBytes(data.get(4)));
        return put;
    }

    /**
     * Insert data
     */
    public void testPut() {
        log.info("Entering testPut.");

        // Specify the column family name.
        byte[] familyName = Bytes.toBytes("info");
        // Specify the column name.
        byte[][] qualifiers = {
                Bytes.toBytes("name"), Bytes.toBytes("gender"), Bytes.toBytes("age"), Bytes.toBytes("address")
        };

        Table table = null;
        try {
            // Instantiate an HTable object.
            table = hbaseConnection.getTable(tableName);
            List<Put> puts = new ArrayList<Put>();

            // Instantiate a Put object.
            Put put = putData(familyName, qualifiers,
                    Arrays.asList("012005000201", "Zhang San", "Male", "19", "Shenzhen, Guangdong"));
            puts.add(put);

            put = putData(familyName, qualifiers,
                    Arrays.asList("012005000202", "Li Wanting", "Female", "23", "Shijiazhuang, Hebei"));
            puts.add(put);

            put = putData(familyName, qualifiers,
                    Arrays.asList("012005000203", "Wang Ming", "Male", "26", "Ningbo, Zhejiang"));
            puts.add(put);

            put = putData(familyName, qualifiers,
                    Arrays.asList("012005000204", "Li Gang", "Male", "18", "Xiangyang, Hubei"));
            puts.add(put);

            put = putData(familyName, qualifiers,
                    Arrays.asList("012005000205", "Zhao Enru", "Female", "21", "Shangrao, Jiangxi"));
            puts.add(put);

            put = putData(familyName, qualifiers,
                    Arrays.asList("012005000206", "Chen Long", "Male", "32", "Zhuzhou, Hunan"));
            puts.add(put);

            put = putData(familyName, qualifiers,
                    Arrays.asList("012005000207", "Zhou Wei", "Female", "29", "Nanyang, Henan"));
            puts.add(put);

            put = putData(familyName, qualifiers,
                    Arrays.asList("012005000208", "Yang Yiwen", "Female", "30", "Kaixian, Chongqing"));
            puts.add(put);

            put = putData(familyName, qualifiers,
                    Arrays.asList("012005000209", "Xu Bing", "Male", "26", "Weinan, Shaanxi"));
            puts.add(put);

            put = putData(familyName, qualifiers,
                    Arrays.asList("012005000210", "Xiao Kai", "Male", "25", "Dalian, Liaoning"));
            puts.add(put);

            // Submit a put request.
            table.put(puts);

            log.info("Put successfully.");
        } catch (IOException e) {
            log.error("Put failed ", e);
        } finally {
            if (table != null) {
                try {
                    // Close the HTable object.
                    table.close();
                } catch (IOException e) {
                    log.error("Close table failed ", e);
                }
            }
        }
        log.info("Exiting testPut.");
    }

    /**
     * createIndex
     */
    public void createIndex() {
        log.info("Entering createIndex.");

        String indexName = "index_name";

        // Create hindex instance
        TableIndices tableIndices = new TableIndices();
        HIndexSpecification iSpec = new HIndexSpecification(indexName);
        iSpec.addIndexColumn(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("info")).build(),
                "name", HIndexProtos.ColumnQualifier.ValueType.STRING);
        tableIndices.addIndex(iSpec);

        HIndexAdmin iAdmin = null;
        Admin admin = null;
        try {
            admin = hbaseConnection.getAdmin();
            iAdmin = HIndexClient.newHIndexAdmin(admin);

            // add index to the table
            iAdmin.addIndices(tableName, tableIndices);
            log.info("Create index successfully.");

        } catch (IOException e) {
            log.error("Create index failed.", e);
        } finally {
            if (admin != null) {
                try {
                    admin.close();
                } catch (IOException e) {
                    log.error("Close admin failed ", e);
                }
            }
            if (iAdmin != null) {
                try {
                    // Close IndexAdmin Object
                    iAdmin.close();
                } catch (IOException e) {
                    log.error("Close admin failed ", e);
                }
            }
        }
        log.info("Exiting createIndex.");
    }

    /**
     * enableIndex
     */
    public void enableIndex() {
        log.info("Entering createIndex.");

        // Name of the index to be enabled
        String indexName = "index_name";

        List<String> indexNameList = new ArrayList<String>();
        indexNameList.add(indexName);

        HIndexAdmin iAdmin = null;
        Admin admin = null;
        try {
            admin = hbaseConnection.getAdmin();
            iAdmin = HIndexClient.newHIndexAdmin(admin);

            // Alternately, enable the specified indices
            iAdmin.enableIndices(tableName, indexNameList);
            log.info("Successfully enable indices {}  of the table {}", indexNameList, tableName);
        } catch (IOException e) {
            log.error("Failed to enable indices {}  of the table {} . {}", indexNameList, tableName, e);
        } finally {
            if (admin != null) {
                try {
                    admin.close();
                } catch (IOException e) {
                    log.error("Close admin failed ", e);
                }
            }
            if (iAdmin != null) {
                try {
                    iAdmin.close();
                } catch (IOException e) {
                    log.error("Close admin failed ", e);
                }
            }
        }
    }

    /**
     * disableIndex
     */
    public void disableIndex() {
        log.info("Entering disableIndex.");

        // Name of the index to be disabled
        String indexName = "index_name";

        List<String> indexNameList = new ArrayList<String>();
        indexNameList.add(indexName);

        HIndexAdmin iAdmin = null;
        Admin admin = null;
        try {
            admin = hbaseConnection.getAdmin();
            iAdmin = HIndexClient.newHIndexAdmin(admin);

            // Alternately, enable the specified indices
            iAdmin.disableIndices(tableName, indexNameList);
            log.info("Successfully disable indices {}  of the table {}", indexNameList, tableName);
        } catch (IOException e) {
            log.error("Failed to disable indices {}  of the table {} . {}", indexNameList, tableName, e);
        } finally {
            if (admin != null) {
                try {
                    admin.close();
                } catch (IOException e) {
                    log.error("Close admin failed ", e);
                }
            }
            if (iAdmin != null) {
                try {
                    iAdmin.close();
                } catch (IOException e) {
                    log.error("Close admin failed ", e);
                }
            }
        }
    }

    /**
     * Scan data by secondary index.
     */
    public void testScanDataByIndex() {
        log.info("Entering testScanDataByIndex.");

        Table table = null;
        ResultScanner scanner = null;
        try {
            table = hbaseConnection.getTable(tableName);

            // Create a filter for indexed column.
            Filter filter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("name"),
                    CompareOperator.EQUAL, "Li Gang".getBytes());
            Scan scan = new Scan();
            scan.setFilter(filter);
            scanner = table.getScanner(scan);
            log.info("Scan indexed data.");

            for (Result result : scanner) {
                for (Cell cell : result.rawCells()) {
                    log.info("{}:{},{},{}", Bytes.toString(CellUtil.cloneRow(cell)),
                            Bytes.toString(CellUtil.cloneFamily(cell)), Bytes.toString(CellUtil.cloneQualifier(cell)),
                            Bytes.toString(CellUtil.cloneValue(cell)));
                }
            }
            log.info("Scan data by index successfully.");
        } catch (IOException e) {
            log.error("Scan data by index failed ", e);
        } finally {
            if (scanner != null) {
                // Close the scanner object.
                scanner.close();
            }
            try {
                if (table != null) {
                    table.close();
                }
            } catch (IOException e) {
                log.error("Close table failed ", e);
            }
        }
        log.info("Exiting testScanDataByIndex.");
    }

    /**
     * Modify a Table
     */
    public void testModifyTable() {
        log.info("Entering testModifyTable.");

        // Specify the column family name.
        byte[] familyName = Bytes.toBytes("education");

        Admin admin = null;
        try {
            // Instantiate an Admin object.
            admin = hbaseConnection.getAdmin();

            // Obtain the table descriptor.
            TableDescriptor htd = admin.getDescriptor(tableName);

            // Check whether the column family is specified before modification.
            if (!htd.hasColumnFamily(familyName)) {
                // Create the column descriptor.
                TableDescriptor tableBuilder = TableDescriptorBuilder.newBuilder(htd)
                        .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(familyName).build())
                        .build();

                // Disable the table to get the table offline before modifying
                // the table.
                admin.disableTable(tableName);
                // Submit a modifyTable request.
                admin.modifyTable(tableBuilder);
                // Enable the table to get the table online after modifying the
                // table.
                admin.enableTable(tableName);
            }
            log.info("Modify table successfully.");
        } catch (IOException e) {
            log.error("Modify table failed ", e);
        } finally {
            if (admin != null) {
                try {
                    // Close the Admin object.
                    admin.close();
                } catch (IOException e) {
                    log.error("Close admin failed ", e);
                }
            }
        }
        log.info("Exiting testModifyTable.");
    }

    /**
     * Get Data
     */
    public void testGet() {
        log.info("Entering testGet.");

        // Specify the column family name.
        byte[] familyName = Bytes.toBytes("info");
        // Specify the column name.
        byte[][] qualifier = {Bytes.toBytes("name"), Bytes.toBytes("address")};
        // Specify RowKey.
        byte[] rowKey = Bytes.toBytes("012005000201");

        Table table = null;
        try {
            // Create the Configuration instance.
            table = hbaseConnection.getTable(tableName);

            // Instantiate a Get object.
            Get get = new Get(rowKey);

            // Set the column family name and column name.
            get.addColumn(familyName, qualifier[0]);
            get.addColumn(familyName, qualifier[1]);

            // Submit a get request.
            Result result = table.get(get);

            // Print query results.
            for (Cell cell : result.rawCells()) {
                log.info("{}:{},{},{}", Bytes.toString(CellUtil.cloneRow(cell)),
                        Bytes.toString(CellUtil.cloneFamily(cell)), Bytes.toString(CellUtil.cloneQualifier(cell)),
                        Bytes.toString(CellUtil.cloneValue(cell)));
            }
            log.info("Get data successfully.");
        } catch (IOException e) {
            log.error("Get data failed ", e);
        } finally {
            if (table != null) {
                try {
                    // Close the HTable object.
                    table.close();
                } catch (IOException e) {
                    log.error("Close table failed ", e);
                }
            }
        }
        log.info("Exiting testGet.");
    }

    /**
     * testScanData
     */
    public void testScanData() {
        log.info("Entering testScanData.");

        Table table = null;
        // Instantiate a ResultScanner object.
        ResultScanner rScanner = null;
        try {
            // Create the Configuration instance.
            table = hbaseConnection.getTable(tableName);

            // Instantiate a Get object.
            Scan scan = new Scan();
            scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"));

            // Set the cache size.
            scan.setCaching(1000);

            // Submit a scan request.
            rScanner = table.getScanner(scan);

            // Print query results.
            for (Result r = rScanner.next(); r != null; r = rScanner.next()) {
                for (Cell cell : r.rawCells()) {
                    log.info("{}:{},{},{}", Bytes.toString(CellUtil.cloneRow(cell)),
                            Bytes.toString(CellUtil.cloneFamily(cell)), Bytes.toString(CellUtil.cloneQualifier(cell)),
                            Bytes.toString(CellUtil.cloneValue(cell)));
                }
            }
            log.info("Scan data successfully.");
        } catch (IOException e) {
            log.error("Scan data failed ", e);
        } finally {
            if (rScanner != null) {
                // Close the scanner object.
                rScanner.close();
            }
            if (table != null) {
                try {
                    // Close the HTable object.
                    table.close();
                } catch (IOException e) {
                    log.error("Close table failed ", e);
                }
            }
        }
        log.info("Exiting testScanData.");
    }

    /**
     * testSingleColumnValueFilter
     */
    public void testSingleColumnValueFilter() {
        log.info("Entering testSingleColumnValueFilter.");

        Table table = null;

        // Instantiate a ResultScanner object.
        ResultScanner rScanner = null;

        try {
            // Create the Configuration instance.
            table = hbaseConnection.getTable(tableName);

            // Instantiate a Get object.
            Scan scan = new Scan();
            scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"));

            // Set the filter criteria.
            SingleColumnValueFilter filter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("name"),
                    CompareOperator.EQUAL, Bytes.toBytes("Xu Bing"));

            scan.setFilter(filter);

            // Submit a scan request.
            rScanner = table.getScanner(scan);

            // Print query results.
            for (Result r = rScanner.next(); r != null; r = rScanner.next()) {
                for (Cell cell : r.rawCells()) {
                    log.info("{}:{},{},{}", Bytes.toString(CellUtil.cloneRow(cell)),
                            Bytes.toString(CellUtil.cloneFamily(cell)), Bytes.toString(CellUtil.cloneQualifier(cell)),
                            Bytes.toString(CellUtil.cloneValue(cell)));
                }
            }
            log.info("Single column value filter successfully.");
        } catch (IOException e) {
            log.error("Single column value filter failed ", e);
        } finally {
            if (rScanner != null) {
                // Close the scanner object.
                rScanner.close();
            }
            if (table != null) {
                try {
                    // Close the HTable object.
                    table.close();
                } catch (IOException e) {
                    log.error("Close table failed ", e);
                }
            }
        }
        log.info("Exiting testSingleColumnValueFilter.");
    }

    /**
     * testFilterList
     */
    public void testFilterList() {
        log.info("Entering testFilterList.");

        Table table = null;

        // Instantiate a ResultScanner object.
        ResultScanner rScanner = null;

        try {
            // Create the Configuration instance.
            table = hbaseConnection.getTable(tableName);

            // Instantiate a Get object.
            Scan scan = new Scan();
            scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"));

            // Instantiate a FilterList object in which filters have "and"
            // relationship with each other.
            FilterList list = new FilterList(FilterList.Operator.MUST_PASS_ALL);
            // Obtain data with age of greater than or equal to 20.
            list.addFilter(new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("age"),
                    CompareOperator.GREATER_OR_EQUAL,
                    Bytes.toBytes(Long.valueOf(20))));
            // Obtain data with age of less than or equal to 29.
            list.addFilter(
                    new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("age"), CompareOperator.LESS_OR_EQUAL,
                            Bytes.toBytes(Long.valueOf(29))));

            scan.setFilter(list);

            // Submit a scan request.
            rScanner = table.getScanner(scan);
            // Print query results.
            for (Result r = rScanner.next(); r != null; r = rScanner.next()) {
                for (Cell cell : r.rawCells()) {
                    log.info("{}:{},{},{}", Bytes.toString(CellUtil.cloneRow(cell)),
                            Bytes.toString(CellUtil.cloneFamily(cell)), Bytes.toString(CellUtil.cloneQualifier(cell)),
                            Bytes.toString(CellUtil.cloneValue(cell)));
                }
            }
            log.info("Filter list successfully.");
        } catch (IOException e) {
            log.error("Filter list failed ", e);
        } finally {
            if (rScanner != null) {
                // Close the scanner object.
                rScanner.close();
            }
            if (table != null) {
                try {
                    // Close the HTable object.
                    table.close();
                } catch (IOException e) {
                    log.error("Close table failed ", e);
                }
            }
        }
        log.info("Exiting testFilterList.");
    }

    /**
     * deleting data
     */
    public void testDelete() {
        log.info("Entering testDelete.");

        byte[] rowKey = Bytes.toBytes("012005000201");

        Table table = null;
        try {
            // Instantiate an HTable object.
            table = hbaseConnection.getTable(tableName);

            // Instantiate an Delete object.
            Delete delete = new Delete(rowKey);

            // Submit a delete request.
            table.delete(delete);

            log.info("Delete table successfully.");
        } catch (IOException e) {
            log.error("Delete table failed ", e);
        } finally {
            if (table != null) {
                try {
                    // Close the HTable object.
                    table.close();
                } catch (IOException e) {
                    log.error("Close table failed ", e);
                }
            }
        }
        log.info("Exiting testDelete.");
    }

    /**
     * dropIndex
     */
    public void dropIndex() {
        log.info("Entering dropIndex.");

        String indexName = "index_name";
        List<String> indexNameList = new ArrayList<>();
        indexNameList.add(indexName);

        HIndexAdmin iAdmin = null;
        try {
            // Instantiate HIndexAdmin Object
            iAdmin = HIndexClient.newHIndexAdmin(hbaseConnection.getAdmin());

            // Delete Secondary Index
            iAdmin.dropIndices(tableName, indexNameList);

            log.info("Drop index successfully.");
        } catch (IOException e) {
            log.error("Drop index failed ", e);
        } finally {
            if (iAdmin != null) {
                try {
                    // Close Secondary Index
                    iAdmin.close();
                } catch (IOException e) {
                    log.error("Close admin failed ", e);
                }
            }
        }
        log.info("Exiting dropIndex.");
    }

    /**
     * Delete user table
     */
    public void dropTable() {
        log.info("Entering dropTable.");

        Admin admin = null;
        try {
            admin = hbaseConnection.getAdmin();
            if (admin.tableExists(tableName)) {
                // Disable the table before deleting it.
                admin.disableTable(tableName);

                // Delete table.
                admin.deleteTable(tableName);
            }
            log.info("Drop table successfully.");
        } catch (IOException e) {
            log.error("Drop table failed ", e);
        } finally {
            if (admin != null) {
                try {
                    // Close the Admin object.
                    admin.close();
                } catch (IOException e) {
                    log.error("Close admin failed ", e);
                }
            }
        }
        log.info("Exiting dropTable.");
    }

    /**
     * grantACL
     */
    public void grantACL() {
        log.info("Entering grantACL.");

        String user = "huawei";
        String permissions = "RW";

        String familyName = "info";
        String qualifierName = "name";

        Table mt = null;
        Admin hAdmin = null;
        try {
            // Create ACL Instance
            mt = hbaseConnection.getTable(PermissionStorage.ACL_TABLE_NAME);

            Permission perm = new Permission(Bytes.toBytes(permissions));

            hAdmin = hbaseConnection.getAdmin();
            TableDescriptor ht = hAdmin.getDescriptor(tableName);

            // Judge whether the table exists
            if (hAdmin.tableExists(mt.getName())) {
                // Judge whether ColumnFamily exists
                if (ht.hasColumnFamily(Bytes.toBytes(familyName))) {
                    // grant permission
                    AccessControlClient.grant(hbaseConnection, tableName, user, Bytes.toBytes(familyName),
                            (qualifierName == null ? null : Bytes.toBytes(qualifierName)), perm.getActions());
                } else {
                    // grant permission
                    AccessControlClient.grant(hbaseConnection, tableName, user, null, null, perm.getActions());
                }
            }
            log.info("Grant ACL successfully.");
        } catch (Throwable e) {
            log.error("Grant ACL failed ", e);
        } finally {
            if (mt != null) {
                try {
                    // Close
                    mt.close();
                } catch (IOException e) {
                    log.error("Close table failed ", e);
                }
            }

            if (hAdmin != null) {
                try {
                    // Close Admin Object
                    hAdmin.close();
                } catch (IOException e) {
                    log.error("Close admin failed ", e);
                }
            }
        }
        log.info("Exiting grantACL.");
    }
}
